Repository navigation
feat(layout): compile segment scans, filters, concatenations, and packs to pipelines - #10392
Merged
Merged
Conversation
Merging this PR will degrade performance by 26.14%
|
| Mode | Benchmark | BASE |
HEAD |
Efficiency | |
|---|---|---|---|---|---|
| ❌ | Simulation | decode_primitives[i64, (1000, 32)] |
24.7 µs | 38.2 µs | -35.4% |
| ❌ | WallTime | threshold_neon[80] |
4.7 µs | 6.6 µs | -29.59% |
| ❌ | WallTime | threshold_neon[8] |
2.9 µs | 4 µs | -27.6% |
| ❌ | WallTime | threshold_neon[64] |
4.3 µs | 5.9 µs | -25.77% |
| ❌ | WallTime | bitpack_blocked_decompress_neon[u32] |
13.6 µs | 15.1 µs | -10.1% |
Tip
Investigate this regression by commenting @codspeedbot fix this regression on this PR, or directly use the CodSpeed MCP with your agent.
Comparing ji/exec-3-scan-concat-pack-nodes (1386a33) with develop (e277ccd)1
Footnotes
joseph-isaacs
added this pull request to stack #10395
October 8, 2026 16:01
Add `Filter`, a plan operator with no predicate that keeps only the rows of the selection it is executed with. Its child produces every row of the same row domain, and values outside the selection are unspecified. Flat layouts now lower to a `Filter` over their `SegmentScan`, so a bare scan can stay dense and return every row of its range while the filter above it keeps the selected ones. Plan display snapshots are updated for the extra level. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Add `plan::exec`, a push-based executor interface for physical plans. A node is built by `PlanVTable::exec` for a range of its plan's rows and a selection over them. It has input ports, one output, and two entry points: `start`, called once to spawn children and request segments, and `compute`, called whenever something changed for the node and its readiness rule holds. A port is an ordered queue of arrays fed by one child plus a closed flag, owned by the graph and read by the node through `StepCx::input`. Every node emits its selected rows in row order as non-empty arrays and returns `Done` to close its own output; there are no close events, no empty arrays, and no row ranges on the data. `Ready` is the readiness rule: `Any` for nodes that stream input through, `Closed(ports)` for nodes that need one input whole before using the others, as a dictionary needs its values before its codes, and `AllClosed` for nodes that need everything. Chains of single-port `Any` nodes are pipelines; the other rules are the barriers between them. `ExecGraph` is the flat node arena. It wires ports, routes segment deliveries to the leaves that requested them, runs ready nodes last-in first-out so data drains toward the root, and hands its owner the root's arrays and batches of `IoRequest`s. IO may complete in any order; a node whose children are backed by independent reads holds the early ones in their ports and emits the ordered prefix. `plan::exec::synthetic` has plans whose nodes are built by hand, so the graph can be exercised without layouts. The scheduling tests drive sources that cut their rows into pieces, request in-memory segments, and yield, under a probe node that records when its readiness rule let it run. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Implement `PlanVTable::exec` for the operators a projection of a struct of chunked columns needs: - `SegmentScanNode` requests its segment at start, decodes the delivered bytes, and emits the node's rows as one dense array. A `Filter` over a scan fuses into the same node with a filter mask, so the array holds exactly the selected rows. Decoded segments are shared through the graph's `DecodeCache`. - `FilterNode` keeps the selected rows of any other child's dense arrays as they stream through, tracking its position with a cursor. - `ConcatNode` spawns only the chunks overlapping its rows, one port each in chunk order, and emits the current chunk's rows as they arrive. Chunks whose reads land early wait in their ports. - `PackNode` zips its fields as their rows arrive: whenever every port has rows, it takes as many as the shortest holds from each and emits one struct. Fields chunked alike stream one struct per chunk. The layout tests drive lowered plans over in-memory segments with scripted IO ordering and check what reaches the root. The scheduling tests add hand-built sources under the real operators, including random trees over random views with reads completing in random order, checked against a model of the same tree. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
joseph-isaacs
force-pushed
the
ji/exec-3-scan-concat-pack-nodes
branch
from
October 8, 2026 20:16
11bd60d to
c31fbcf
Compare
Replaces the `ExecNode` graph with `plan::pipeline`, the executor a plan now runs on: - `Operator` and `Source`: one `compute` method returning a `Step` (`More`, `Last`, `Consumed`, `Blocked`, `Finished`). Only a source blocks, on a read, an empty inlet, or a full outlet. - Ports with one writer and one reader, in an arena the scan owns; the reading source declares each inlet's capacity, and `Inlet` exposes `peek_mut`, `take`, and `closed`. - A driver that pushes each batch through a pipeline's stages into its outlets, and a scheduler that wakes a blocked pipeline only when its condition clears, checking only the ports a run touched. - `Scan`, which compiles a plan per split, runs several splits at once, and returns reads to its owner as `Turn::Read`. - `PlanVTable::compile` and `PlanVTable::reach` replace `PlanVTable::exec`: the hooks a plan implements to compile into a chain and to report the segments it reads. Synthetic plans exercise the runtime without layouts. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…ks to pipelines Merges the pipeline runtime from ji/exec-2-exec-node and replaces this branch's exec nodes with compile rules on the plans: - SegmentScan reads and decodes one segment; a Filter over it keeps the selected rows in the same source, and over any other child keeps them in a stage per batch. - Concat builds only the chunks a split's mask selects a row of, and joins them with a source that drains its inlets in order; rows inside one chunk are that chunk's rows. - Pack zips its fields' streams into structs, slicing at the boundaries the fields share; a single field is wrapped as it passes. - A segment more than one reader reads is decoded once: one pass over the plan records the rows each segment is read for, the first reader builds one decoding pipeline that fans out into a port per reader, tagged by split, and a split's unclaimed ports go when it finishes. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…ep its own rows A flat layout lowers to a bare segment scan again, as on develop. Every plan now produces only the rows its split selects, a segment scan included, so the scan's source applies the mask; a Filter over a scan compiles to the same source. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
joseph-isaacs
changed the title
`Cx::spawn` let a source ask for a plan to be compiled into a new inlet while the scan ran. Every plan is now compiled up front: what a stage reads is decided by its plan and its masks before it runs, so the runtime never plans mid-run. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…et a reader waits on Ports allocate their buffer once, at their capacity, and keep it when a freed slot is reused. The default capacity rises from 2 to 8 batches, which halves scheduler round trips on narrow pipelines. A stage's output port is bounded like any other, so it needs no growth either; only a share's port, whose reader may be compiled later, can grow past the default. A reader blocked on inlet i is woken only by a batch or close on that inlet, and the inlets a run touched are tracked with a flag per inlet, so a source reading thousands of inlets pays per inlet it read, not per wake. Pipelines and ports live in a slab that never moves its entries. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…akes Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…y boundary Pack used to emit a struct as soon as every field had a front batch, cut to the shortest one. Fields chunked differently were sliced at every boundary of every other field: 10,000 misaligned columns made 8,192 structs and 80 million slices per scan. Now, when every field's front batch has the same length, Pack takes those batches whole, so aligned fields still stream one struct per chunk with no slicing. Otherwise it waits on the one field with the fewest queued rows until that field is full or closed, then takes that many rows from every field. Whole batches are joined as a ChunkedArray, not copied, and at most one batch per field is sliced. Memory stays bounded by the inlet capacities. Waiting costs O(1) per wake. Queued rows only grow until a struct is built, so Pack resumes at the first inlet it found empty instead of rescanning, and while filling it checks only the inlet it waits on. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
A source could ask for one segment at a time: the driver issued its read and did not compute the source until the bytes came back, so a source reading several segments waited for each in turn. Now the driver asks the source for segments until it has none to ask for and issues every read at once. The source is computed as each arrives and takes the bytes tagged with their segment through Cx::take_segment, oldest arrival first. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
The scan's docs linked QueryPlan, which a later change adds, and failed the docs build with broken intra-doc links denied. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Ports are named by their ends: a pipeline's sink pushes into its outlets
and another pipeline's source reads them as inlets (SourceEnd replaces
Reader). Source and Stage replace Operator and Input: a source returns
Step::{Batch, Blocked, Finished}, a stage pushes one batch and flushes on
finish, and the driver owns full outlets. ConcatSource::new(1) replaces
PortSource, and OnceSource moves to its own module.
The driver now:
- cancels the reads a retiring pipeline left in flight, so their bytes
are dropped instead of reaching a pipeline that reuses its id;
- reports Done once no split is live, whatever reads are left over;
- finishes stages one at a time behind the outlets' room check, so a
flush never overfills a port, and closes at once when nothing is left.
Debug builds validate the graph after every compile (port ends agree,
sources read the inlets they declare, every port has a reader, fan-out
is unbounded so no diamond can deadlock) and each run's report (a source
waits only on an open inlet with room, or on a read in flight, and
finishes only once its inlets close). "Scan is stuck" names the port and
sink each waiting source depends on.
Signed-off-by: Claude <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…g validation
Carries the same change into this branch's operators:
Ports are named by their ends: a pipeline's sink pushes into its outlets
and another pipeline's source reads them as inlets (SourceEnd replaces
Reader). Source and Stage replace Operator and Input: a source returns
Step::{Batch, Blocked, Finished}, a stage pushes one batch and flushes on
finish, and the driver owns full outlets. ConcatSource::new(1) replaces
PortSource, and OnceSource moves to its own module.
The driver now:
- cancels the reads a retiring pipeline left in flight, so their bytes
are dropped instead of reaching a pipeline that reuses its id;
- reports Done once no split is live, whatever reads are left over;
- finishes stages one at a time behind the outlets' room check, so a
flush never overfills a port, and closes at once when nothing is left.
Debug builds validate the graph after every compile (port ends agree,
sources read the inlets they declare, every port has a reader, fan-out
is unbounded so no diamond can deadlock) and each run's report (a source
waits only on an open inlet with room, or on a read in flight, and
finishes only once its inlets close). "Scan is stuck" names the port and
sink each waiting source depends on.
Signed-off-by: Claude <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…only Queue::unbounded and Queue::is_free, and the test sources that break the scheduling contract, are used only under debug_assertions, so release builds (cargo check --release in CI) rejected them as dead code. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…ebug builds only Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
… Input Stage::push and Stage::finish become one Operator::compute, handed Input::Batch for each batch and Input::End once, last. The driver still hands End one operator at a time behind the outlets' room check, and an operator still never blocks. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Ports this branch's operators to the same trait. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Source is now one method, ready(), returning Step::{Batch, Read,
Blocked, Finished}. A source asks for a segment by returning
Step::Read; the driver issues the read once the run ends and asks the
source again at once, so it can still have several reads in flight
before it answers Blocked::Io. Bytes already there, as of a segment read
ahead, run the source again at once.
The inlet count comes from the chain, and capacities are the joining
plan's: Compiler::join holds DEFAULT_CAPACITY batches per inlet, and
Compiler::join_with takes a capacity per inlet.
Signed-off-by: Claude <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Ports this branch's sources to Source::ready and Step::Read, and its joins to Compiler::join_with. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…uations Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
…nd compute Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
…' into ji/exec-3-scan-concat-pack-nodes Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
robert3005
approved these changes
Oct 12, 2026
Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Compiles segment scans, filters, concatenations, and packs into runnable pipelines, including shared segment decoding across readers and splits. Tracking: #10349. Stacked on #10391; followed by #10393.
ready()and decode/select rows incompute(). Filters and single-field struct wrappers returnFinal(batch)from the shared operator API.ready()and consumes batches incompute(). Chunks with no selected rows are never compiled or read.Tests cover FIFO/LIFO delivery, misaligned field batches, readiness without rescanning unchanged inlets, consumption updates, mismatched field lengths, skipped chunks, selection, shared decoding, and port cleanup.
Physical plans and their optimizer live under
vortex_layout::plan::physical. The siblingoperatormodule definesOperator,Source, and the execution context;operatorscontains concrete implementations;pipelineowns compilation, scheduling, and ports.Validation:
cargo +nightly-2026-09-10 fmt -p vortex-layout; scopedcargo clippy -p vortex-layout --all-targets --all-features -- -D warnings; targeted physical-plan tests (94 passed).